package com.bd08.flink.demo

import org.apache.flink.api.common.functions.AggregateFunction
import org.apache.flink.streaming.api.scala.{StreamExecutionEnvironment, createTypeInformation}

object TestJdbcSource  extends App {
  val env = StreamExecutionEnvironment.getExecutionEnvironment

  env.addSource(new MyJdbcSource)
    .keyBy(ke => ke.jie).countWindow(3).aggregate(new MyAgg)
    .print()
  env.execute



  // List 问题
  class MyAgg extends AggregateFunction[Ke,String,String]{
    override def createAccumulator(): String = {
      println("createAccumulator")
      "List.empty"
    }

    override def add(in: Ke, acc: String): String = {
      println("add"+in.jie)
      acc +"|"+ in
    }

    override def getResult(acc: String):String = {
      println("getResult")
      "..."
    }

    override def merge(acc:String, acc1: String): String = {
            println("merge")
            acc +"||"+ acc1

    }
  }

}
